package org.example.kafka;

import com.alibaba.fastjson2.JSONObject;
import org.example.dao.repository.ElasticRepository;
import org.example.domain.DiscussPost;
import org.example.domain.Message;
import org.example.service.imp.DiscussServiceImp;
import org.example.service.imp.ElasticSearchServerImp;
import org.example.service.imp.MessageServiceImp;
import org.example.util.CommonParam;
import org.example.util.HutoolUtil;
import org.omg.SendingContext.RunTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;
import java.util.UUID;

@Component
public class MessageConsumer {
    @Autowired
    private MessageServiceImp messageServiceImp;
    @Autowired
    private HutoolUtil hutoolUtil;
    @Autowired
    private ElasticRepository elasticRepository;
    @Autowired
    private ElasticSearchServerImp elasticSearchServerImp;
    @Autowired
    private DiscussServiceImp discussServiceImp;
    @Autowired
    private RedisTemplate<String,Object> redisTemplate;

    private static final Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
    //当前的消费者 消费点赞 关注 和 评论的消息
    @KafkaListener(topics = {CommonParam.KAFKA_TOPIC_LIKE,CommonParam.KAFKA_TOPIC_COMMENT,CommonParam.KAFKA_TOPIC_FOLLOW})
    public void messageHandler(String message){
        if (hutoolUtil.stringIsEmpty(message)){
            logger.error("消息格式内容为空");
            return;
        }
        Event event = JSONObject.parseObject(message, Event.class);
        if (event==null){
            logger.error("消息格式不正确");
            return;
        }
        //所谓的consumer处理消息 就是将当前的信息封装成Message储存mysql
        Message result = new Message();
        //如果当前是一个点赞的消息处理
        //某个人给你的(帖子或者评论点赞)  系统通知你  某个用户点赞了你的(帖子或者评论)  点赞者的id
        result.setFromId(1);//默认系统用户为1   因为这个不会作为查询条件 conversation_id才是两两信息的查询条件
        result.setToId(event.getEntityUserId());
        String conversationId = "system_"+event.getTopic()+"_"+event.getEntityUserId();
        result.setConversationId(conversationId);
        result.setContent(message);
        result.setStatus((byte)1);
        result.setCreateTime(new Date());
        messageServiceImp.saveOne(result);

    }

    @KafkaListener(topics = {CommonParam.KAFKA_TOPIC_DISCUSS_POST})
    public void handleDiscuss(String message){
        if (hutoolUtil.stringIsEmpty(message)){
            logger.info(this.getClass().toString()+"：接受的message为null");
            return;
        }
        DiscussPost discussPost = JSONObject.parseObject(message,DiscussPost.class);
        if (discussPost==null){
            logger.info(this.getClass().toString()+"：接受的message格式有问题");
            return;
        }
        discussServiceImp.saveOne(discussPost);
        elasticRepository.save(discussPost);
    }
    @KafkaListener(topics = {CommonParam.KAFKA_TOPIC_UPDATE_DISCUSS})
    public void updateDiscuss(String message){
        if (hutoolUtil.stringIsEmpty(message)){
            logger.info(this.getClass().toString()+"：接受的message为null");
            return;
        }
        DiscussPost discussPost = JSONObject.parseObject(message,DiscussPost.class);
        if (discussPost==null){
            logger.info(this.getClass().toString()+"：接受的message格式有问题");
            return;
        }

        elasticRepository.save(discussPost);
    }
    @KafkaListener(topics = {CommonParam.KAFKA_TOPIC_DELETE_DISCUSS})
    public void deleteDiscuss(Integer id){
        if (id==null || id<=0){
            logger.info("需要删除的帖子id为null");
            return;
        }
        elasticRepository.deleteById(id);
    }
    @Value("${wk.image.cmd}")
    private String generateImageCmd;
    @Value("${wk.generate.image.location}")
    private String systemWkGenerateImageLocation;
    @KafkaListener(topics = {CommonParam.KAFKA_TOPIC_WK_GENERATE_IMAGE})
    public void generateWkImage(String wkEvent){

        if (hutoolUtil.stringIsEmpty(wkEvent)){
            logger.info("kafka处理"+CommonParam.KAFKA_TOPIC_WK_GENERATE_IMAGE+"主题失败!!原因：消息为空");
            return;
        }
        WKEvent wkEvent1 = JSONObject.parseObject(wkEvent, WKEvent.class);
        if (wkEvent1 == null || hutoolUtil.stringIsEmpty(wkEvent1.getWebLocation()) || wkEvent1.getUserId()==null){
            logger.info("kafka处理"+CommonParam.KAFKA_TOPIC_WK_GENERATE_IMAGE+"主题失败!!原因：消息格式不正确");
            return;
        }

        StringBuilder cmd  = new StringBuilder();
        cmd.append(generateImageCmd);
        cmd.append(" ");
        cmd.append(wkEvent1.getWebLocation());//这个网页可能存在问题 格式问题
        cmd.append(" ");
        cmd.append(systemWkGenerateImageLocation);
        //添加图片的具体名字
        String imageName = hutoolUtil.randomString(6);
        cmd.append(imageName);
        cmd.append(".png");
        //我们需要将生产的图片 传给用户 文件上传  需要一个文件的名字  我们可以将这个文件名字存储redis中
        redisTemplate.opsForSet().add(CommonParam.REDIS_WK_GENERATE_IMAGE_SYMBOL + wkEvent1.getUserId(),imageName+".png");

        try {
            Runtime.getRuntime().exec(cmd.toString());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
